/*
 * Hyuga: A Density-Grid Stream Clustering Platform.
 * Copyright (C) 2014 PUC-Rio/Laboratory for Advanced Collaboration
 *
 * Hyuga is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Hyuga is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Hyuga.  If not, see <http://www.gnu.org/licenses/>.
 */
package br.pucrio.inf.lac.konoha.hyuga.listeners;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import br.pucrio.inf.lac.konoha.hyuga.core.DDSLink;
import br.pucrio.inf.lac.konoha.hyuga.events.LocationUpdateEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.csv.CSVLocationUpdateEvent;

/**
 * A listener class that receives CSV Location Updates and output a {@link LocationUpdateEvent} to the SDDL Domain and
 * outside applications (monitoring, benchmark, etc).
 * 
 * @author Marcos Roriz <mroriz@inf.puc-rio.br>
 */
public class CSVListener {
    /** Data Stream. */
    private DDSLink dataStream;
	
    /** Thread Pool for sending events. */
    protected ExecutorService sendPool;
    
	public CSVListener(DDSLink dataStream) {
		this.dataStream	= dataStream;
        this.sendPool   = Executors.newFixedThreadPool(16);
	}
	
	public void update(CSVLocationUpdateEvent csvEvent) {
		sendPool.submit(new SendCSVLocationUpdate(csvEvent));
	}
	
	private class SendCSVLocationUpdate implements Runnable {
	    private LocationUpdateEvent locationUpdateEvent;
	    
	    public SendCSVLocationUpdate(CSVLocationUpdateEvent csvEvent) {
            locationUpdateEvent = new LocationUpdateEvent(csvEvent);
        }

        @Override
        public void run() {
            dataStream.publishSpatial(locationUpdateEvent.getOutputTopic(), 
                                      locationUpdateEvent.getLatitude()  * 1000000,
                                      locationUpdateEvent.getLongitude() * 1000000, 
                                      locationUpdateEvent.getType().toString(), 
                                      locationUpdateEvent,
                                      locationUpdateEvent.getType().toString());     
        }
	}
}
